package org.unige.mpej.eckmann.sonedyan.hama;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;

import org.apache.hama.HamaConfiguration;
import org.apache.hama.bsp.BSP;
import org.apache.hama.bsp.BSPJobClient;
import org.apache.hama.bsp.BSPPeer;
import org.apache.hama.bsp.ClusterStatus;
import org.apache.hama.bsp.Combiner;
import org.apache.hama.bsp.HashPartitioner;
import org.apache.hama.bsp.TextInputFormat;
import org.apache.hama.bsp.TextOutputFormat;
import org.apache.hama.bsp.sync.SyncException;
import org.apache.hama.graph.AbstractAggregator;
import org.apache.hama.graph.Edge;
import org.apache.hama.graph.GraphJob;
import org.apache.hama.graph.Vertex;
import org.apache.hama.graph.VertexInputReader;
import org.apache.log4j.Logger;

/**
 * find the shortest cycles of a given length
 */
public class ShortestCyclesFinder
{
	private static Logger log = Logger.getLogger(org.unige.mpej.eckmann.sonedyan.hama.ShortestCyclesFinder.class);

	/**
	 * 
	 * <V> this type must be writable and should also implement equals and hashcode.
	 * <E> the type used for storing edge values, usually the value of an edge.
	 * <M> the type used for messaging, usually the value of a vertex.
	 */
	public static class ShortestCyclesVertex extends Vertex<IntWritable, NullWritable, Text>
	{
		// http://hadoop.apache.org/docs/stable/api/index.html
		@Override
		public void setup(Configuration conf)
		{
		}

		@Override
		public void compute(Iterator<Text> messages) throws IOException
		{
			// initialize this vertex to 
			if (this.getSuperstepCount() == 0)
			{
				this.setValue(new Text(Integer.toString(this.getVertexID().get())));
			} 
			else if (this.getSuperstepCount() >= 1) 
			{
				// id1,id2:id3,id2
				String vCycles = "";

				while (messages.hasNext())
				{
					Text msg = messages.next();

					String[] split = msg.toString().split(":");

					// add current vertex id to the set of paths received
					for (int i = 0; i < split.length; i++)
					{
						if (vCycles.length() == 0)
							vCycles += split[i] + "," + this.getVertexID().get();
						else
							vCycles += ":" + split[i] + "," + this.getVertexID().get();
					}
				}

				this.setValue(new Text(vCycles));
			}
			
			if (this.getSuperstepCount() == this.getMaxIteration()) 
			{
				voteToHalt();
				return;
			}

			// in each superstep we are going to send a new rank to our neighbours
			this.sendMessageToNeighbors(this.getValue());
		}
	}

	public static class ShortestCyclesTextReader extends VertexInputReader<LongWritable, Text, IntWritable, NullWritable, Text>
	{
		/**
		 * The text file essentially should look like:
		 * 
		 * VERTEX_ID\t(space separated VERTEX_IDs)
		 * 
		 * E.G:
		 * 1 2 3 4
		 * 2 3 1
		 * ...
		 * 
		 * This should be able to load a iGraph edge list formatted file
		 */
		@Override
		public boolean parseVertex(LongWritable key, Text value, Vertex<IntWritable, NullWritable, Text> vertex) throws Exception 
		{
			String[] split = value.toString().split(" ");
			for (int i = 0; i < split.length; i++)
			{
				if (i == 0)
				{
					vertex.setVertexID(new IntWritable(Integer.parseInt(split[i])));
				} 
				else 
				{
					vertex.addEdge(new Edge<IntWritable, NullWritable>(new IntWritable(Integer.parseInt(split[i])), null));
				}
			}
			return true;
		}
	}

	private static void printUsage()
	{
	  System.out.println("Usage: <input> <output> <cycle_length>");
	  System.exit(-1);
	}

	/**
	 * main method
	 */
	public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException
	{
		if (args.length != 3)
			printUsage();

		HamaConfiguration conf = new HamaConfiguration(new Configuration());
		GraphJob pageJob = createJob(args, conf);
		
		long startTime = System.currentTimeMillis();
		
		if (pageJob.waitForCompletion(true))
		{
			log.info("Job Finished in " + (System.currentTimeMillis() - startTime) / 1000.0 + " seconds");
		}
	}

	@SuppressWarnings("unchecked")
	public static GraphJob createJob(String[] args, HamaConfiguration conf) throws IOException
	{
		GraphJob pageJob = new GraphJob(conf, ShortestCyclesFinder.class);
		pageJob.setJobName("ShortestCycles");

  		pageJob.setVertexClass(ShortestCyclesVertex.class);
  		pageJob.setInputPath(new Path(args[0]));
  		pageJob.setOutputPath(new Path(args[1]));
  		pageJob.setMaxIteration(Integer.parseInt(args[2]));
		
  		//pageJob.setBspClass(ShortestCyclesBSP.class);
	  	pageJob.setAggregatorClass(ShortestCyclesAggregator.class);
	  	
	  	pageJob.setVertexIDClass(IntWritable.class);
	  	pageJob.setVertexValueClass(Text.class);
	  	pageJob.setEdgeValueClass(NullWritable.class);
	
	  	pageJob.setInputKeyClass(LongWritable.class);
	  	pageJob.setInputValueClass(Text.class);
	  	pageJob.setInputFormat(TextInputFormat.class);
	  	pageJob.setVertexInputReaderClass(ShortestCyclesTextReader.class);
	  
	  	pageJob.setPartitioner(HashPartitioner.class);
	  	pageJob.setOutputFormat(TextOutputFormat.class);
	  	pageJob.setOutputKeyClass(Text.class);
	  	pageJob.setOutputValueClass(Text.class);
	  	
	  	BSPJobClient jobClient = new BSPJobClient(conf);
		ClusterStatus cluster = jobClient.getClusterStatus(true);
	    pageJob.setNumBspTask(cluster.getMaxTasks());
	    
	    log.info("Max number of tasks available: " + cluster.getMaxTasks());
	  	
	  	return pageJob;
	}

	/**
	 * aggregator
	 * 
	 * concatenating all cycles found so far
	 */
	public static class ShortestCyclesAggregator extends AbstractAggregator<Text, Vertex<IntWritable, NullWritable, Text>>
	{
		String cycles;

		@Override
		public void aggregate(Vertex<IntWritable, NullWritable, Text> vertex, Text value)
		{
			if (value != null)
			{
				if (cycles == null)
					cycles = value.toString();
				else
					cycles += "|" + value.toString();
			}
		}

		@Override
		public Text getValue()
		{
			// log.info("Aggregated value: " + cycles);
			return new Text(cycles);
		}
	}
	
	/**
	 * BSP 
	 */
	public static class ShortestCyclesBSP extends BSP<IntWritable, NullWritable, Text, Text, Text>
	{
		private String masterTask;
		
		@Override
	    public void setup(BSPPeer<IntWritable, NullWritable, Text, Text, Text> peer) throws IOException
	    {
	      // Choose one as a master
	      this.masterTask = peer.getPeerName(peer.getNumPeers() / 2);
	    }
		
		@Override
		public void bsp(BSPPeer<IntWritable, NullWritable, Text, Text, Text> peer) throws IOException, SyncException, InterruptedException
		{
		}
		
		@Override
		public void cleanup(BSPPeer<IntWritable, NullWritable, Text, Text, Text> peer) throws IOException
		{
		}
	}
	
}
